﻿using System;
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;

namespace Inovout.YunYao.Cattle.WindowsService
{
    /// <summary>
    /// 放牛人，监控队列处理情况。
    /// </summary>
    public class DataDrover
    {
        internal Thread Thread { get; set; }
        private MonitorTask monitorTask;
        public DataDrover(MonitorTask task)
        {
            this.monitorTask = task;
        }
        private DateTime lastDoneDateTime;
        public void Do()
        {
            Dictionary<PersistentTask, QueueManager> queueDictionary = new Dictionary<PersistentTask, QueueManager>();
            foreach (PersistentTask persistentTask in monitorTask.PersistentTasks)
            {
                queueDictionary.Add(persistentTask,
                    new QueueManager(persistentTask.RedisIP, persistentTask.RedisPort, persistentTask.RedisTableId, persistentTask.QueueName));
            }
            lastDoneDateTime = DateTime.Now;
            while (true)
            {
                foreach (var queue in queueDictionary)
                {
                    long queueCount = queue.Value.Count();
                    if (queueCount >= monitorTask.MaxQueueCount)
                    {
                        AlertManager.Send(string.Format("{0}Redis中{1}Queue数量为：{2}，已超过{3}上限值"
                            , queue.Key.RedisIP, queue.Key.QueueName, queueCount.ToString(), monitorTask.MaxQueueCount.ToString()));
                        //由于从头取会与正常冲突，若从后取，又因后面还有数据进来，无法定位移除，故取拷贝到一个新的队列，单独去处理。
                        //由于复制队列已经可以拿到队列数据，故不再由数据黄牛去获取，而直接发给队列数据，备份队列只做为备份使用，处理完之后，立即清空。
                        var queues = queue.Value.CopyTo(monitorTask.HelperQueueName, monitorTask.BatchCount);
                        if (queues.Length > 0)
                        {
                            DataConveyor dataConveyor = new DataConveyor(queue.Key.HBaseIP, queue.Key.HBasePort, queue.Key.HBaseTableNamePrefix);
                            dataConveyor.Push(queues);
                            using (Task task = queue.Value.Remove(monitorTask.HelperQueueName, 0, (int)queueCount))
                            {
                                task.Wait();
                            }
                        }

                        //DataCattle dataCattle = new DataCattle(new PersistentTask
                        //{
                        //    RedisIP = queue.Key.RedisIP,
                        //    RedisPort = queue.Key.RedisPort,
                        //    RedisTableId = queue.Key.RedisTableId,
                        //    QueueName = task.HelperQueueName,
                        //    HBaseIP = queue.Key.HBaseIP,
                        //    HBasePort = queue.Key.HBasePort,
                        //    HBaseTableName = queue.Key.HBaseTableName,
                        //    BatchCount = queue.Key.BatchCount
                        //});
                        //dataCattle.Do(0, task.BatchCount);

                    }
                }
                TimeSpan sleepTimeSpan = new TimeSpan(monitorTask.IntervalTick - (DateTime.Now.Ticks - lastDoneDateTime.Ticks));
                if (sleepTimeSpan.Ticks > 0)
                {
                    Thread.Sleep(sleepTimeSpan);
                }
                lastDoneDateTime = DateTime.Now;
            }
        }
    }
}
